package com.atguigu.gmall.realtime.app.functions;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

/**
 * @Description: 将侧输出流中的数据, 写入到HBase中.
 * @Author: tiancy
 * @Create: 2021/11/26
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //创建连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /*
        {
            "database":"gmall0609",
            "data":{"tm_name":"hh","id":16},
            "sink_table":"dim_base_trademark",
            "type":"insert",
            "table":"base_trademark"
        }
     */
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取维度数据库表名
        String tableName = jsonObj.getString("sink_table");
        //获取维度记录
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");

        String id = dataJsonObj.getString("id"); // 获取当前存入Hbase中的id值. 我需要通过id值,去删除redis中的缓存数据 (订单宽表具体实现 --优化部分,旁路缓存)
        //拼接upsert语句  upsert into 表空间.表(a,b,c) values(aa,bb,cc)
        String sql = generateUpsertSql(tableName, dataJsonObj);

        System.out.println("向phoenix中插入数据的SQL:" + sql);

        PreparedStatement ps = null;
        try {
            //获取数据库操作对象
            ps = conn.prepareStatement(sql);
            //执行SQL语句
            ps.executeUpdate();
            //注意：phoenix数据库   autocommit默认是false，需要手动提交
            conn.commit();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("向phoenix表中插入维度数据失败");
        } finally {
            //释放资源
            if (ps != null) {
                ps.close();
            }
        }
        /** 维度表中数据修改,需要删除redis中做的旁路缓存*/
        //如果对维度数据进行了修改，那么需要将Redis中缓存的数据删掉
        if (jsonObj.getString("type").equals("update")) {
            DimUtil.deleteCached(tableName, id);
        }

    }

    //向维度表中插入数据的sql字符串的拼接   "data":{"tm_name":"hh","id":16}===>dim_base_trademark
    private String generateUpsertSql(String tableName, JSONObject dataJsonObj) {
        Set<String> keys = dataJsonObj.keySet(); // 这里是在Entry中直接拿到的keys的值,是能够保证数据的,并不是一个一个往set中放,最终导致取的时候不一致问题.
        Collection<Object> values = dataJsonObj.values();

        // 这里拼接SQL,使用的是 org.apache.commons.lang3.StringUtils 的一个工具类.
        String sql = "upsert into " + GmallConfig.HBASE_SCHEMA + "." + tableName
                + "(" + StringUtils.join(keys, ",") + ") values('" + StringUtils.join(values, "','") + "')";
        return sql;
    }
}